package SYY.Service

import bean.ResumeMatchData
import org.apache.spark.streaming.dstream.DStream
import util.SparkUtil

object PositionLanguageMatchService {
  import org.apache.spark.sql.functions._
  def analyzeMatch(resumeDataStream: DStream[ResumeMatchData]): Unit = {
    val spark = SparkUtil.takeSpark()
    import spark.implicits._
    resumeDataStream.foreachRDD(resumeRDD => {
      if (!resumeRDD.isEmpty()) {
        val resumeDF = resumeRDD.toDF()

        val matchDF = resumeDF
          // 新增：计算掌握语言的数量（假设语言用逗号分隔）
          .withColumn("language_count",
            size(split(col("programming_languages"), ","))
          )

        val resultDF = matchDF
          // 修改：按筛选结果和语言数量分组
          .groupBy("screening_result", "language_count")
          .agg(count("*") as "数量")

        resultDF.show()
      }
    })
  }
}